package sicnu.cs.ich.common.common.concurrent;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.parser.Feature;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import sicnu.cs.ich.api.common.Constants;
import sicnu.cs.ich.api.domain.dto.BaseHitComparator;
import sicnu.cs.ich.api.common.enums.ConfigEnum;


import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 用于热力值统计
 *
 * @author CaiKe
 * @date 2021/11/8 14:37
 */
@Slf4j
public class HitCalculateQueue<T extends BaseHitComparator<T>> {

    private static final int INITIAL_CAPACITY = Constants.MAX_ICH_COUNT;
    /**
     * 有效值最低阀值
     */
    private final double lowestThreshold;
    /**
     * maximum size of hit-ich stored
     */
    private final int capacity;
    /**
     * count of ich searched
     */
    private final LongAdder ALL_HITS = new LongAdder();
    private final LongAdder CUR_HITS = new LongAdder();
    /**
     * 非遗热度统计计算（每N小时）
     */
    private final ReentrantLock lock = new ReentrantLock(false);

    private final PriorityQueue<T> RANKING;
    private final ConcurrentHashMap<Integer, Integer> INCRE = new ConcurrentHashMap<>();
    private final HashMap<Integer, Integer> RECORD = new HashMap<>();

    public HitCalculateQueue() {
        this(1.0, Constants.MAX_ICH_COUNT, INITIAL_CAPACITY);
    }

    public HitCalculateQueue(double threshold) {
        this(threshold, Constants.MAX_ICH_COUNT, INITIAL_CAPACITY);
    }


    public HitCalculateQueue(Integer maxCount) {
        this(1.0, maxCount, INITIAL_CAPACITY);
    }


    public HitCalculateQueue(Integer maxCount, Integer initialHits) {
        this(1.0, maxCount, initialHits);
    }

    /**
     * 构造方法
     *
     * @param maxCount      热度榜中的最大保存量
     * @param initialHits   初始化总量大小，默认100
     * @param threshold     最低阀值
     */
    public HitCalculateQueue(double threshold, Integer maxCount, Integer initialHits) {
        lowestThreshold = threshold;
        capacity = maxCount;
        ALL_HITS.add(initialHits);
        RANKING = new PriorityQueue<>(capacity);
    }

    /**
     * 热度+1
     *
     * @param key typeId
     */
    public void increment(Integer key) {
        CUR_HITS.increment();
        incrementBy(key, 1);
    }

    /**
     * 为非遗增加score 【此操作不会增加CUR_HIT】
     */
    public void incrementBy(Integer key, int score) {
        INCRE.merge(key, score, Integer::sum);
    }

    /**
     * 热度+1
     *
     * @param keys
     */
    public void increment(List<Integer> keys) {
        CUR_HITS.add(keys.size());
        keys.forEach(this::increment);
    }

    /**
     * 获取热度列表
     */
    @Deprecated
    public List<Integer> getRankingIds() throws InterruptedException {
        List<Integer> ids;
        lock.lockInterruptibly();
        try {
            ids = RANKING.stream().map(BaseHitComparator::getKey).collect(Collectors.toList());
        } finally {
            lock.unlock();
        }
        return ids;
    }

    public PriorityQueue<T> getRankingObjs() {
        final PriorityQueue<T> ts;
        synchronized (RANKING) {
            ts = new PriorityQueue<>(RANKING);
        }
        return ts;
    }

    /**
     * 热度排序
     * @param function 构建T的方法
     * @param getHour 获取与上次更新的时间间隔，小于 1h 则不进行更新
     */
    public void sortRanking(BiFunction<Integer, Integer, T> function, Function<Void, Integer> getHour, Consumer<Void> updateHour) {
        // calculate hour, return when lesser than one
        final Integer gapHour = getHour.apply(null);
        if (gapHour < 1) return;
        ALL_HITS.add(CUR_HITS.longValue());
        double cooling = Math.log((CUR_HITS.doubleValue() + 1) / (ALL_HITS.doubleValue() + 1));
        log.info("开始SORT分数排行榜，本次浏览量：{}，总浏览量：{}，冷却系数：{}", CUR_HITS.longValue(), ALL_HITS.longValue(), cooling);

        // 衰减历史分
        final Iterator<Map.Entry<Integer, Integer>> iteratorRanking = RECORD.entrySet().iterator();
        while (iteratorRanking.hasNext()) {
            val entry = iteratorRanking.next();
            if (entry.getValue() >= lowestThreshold) {
                entry.setValue(entry.getValue() * (int) Math.exp(cooling * gapHour));
            } else {
                iteratorRanking.remove();
            }
        }

        // update UpdatedHour
        updateHour.accept(null);

        lock.lock();
        try {
            CUR_HITS.reset();
            // clear ranking
            RANKING.clear();
            final Iterator<Map.Entry<Integer, Integer>> iterator = INCRE.entrySet().iterator();
            while (iterator.hasNext()) {
                val next = iterator.next();
                final Integer k = next.getKey();
                var newV = next.getKey() + RECORD.getOrDefault(k, 0);
                RECORD.merge(k, newV, Integer::sum);
                // put into ranking
                if (RANKING.size() < capacity) {
                    // insert into ranking
                    RANKING.offer(function.apply(k, newV));
                } else {
                    // compare
                    double hits = RANKING.peek().getHit();
                    if (newV > hits) {
                        RANKING.poll();
                        RANKING.offer(function.apply(k, newV));
                    }
                }
                // clear ConcurrentHashMap
                iterator.remove();
            }
        } finally {
            lock.unlock();
        }
    }


    /**
     * 当前热度榜是否满
     *
     * @return
     */
    public boolean isFull() {
        return RANKING.size() == capacity;
    }

    /**
     * 热度榜最大长度
     */
    public int getCapacity() {
        return capacity;
    }

    public void loadConfig(HashMap<ConfigEnum, String> configMap) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            Optional.ofNullable(configMap.get(ConfigEnum.ICH_HIT_MAP)).ifPresent(value ->
            {
                int disableDecimalFeature = JSON.DEFAULT_PARSER_FEATURE & ~Feature.UseBigDecimal.getMask();
                this.INCRE.putAll(JSON.parseObject(value, ConcurrentHashMap.class, disableDecimalFeature));
            });
            Optional.ofNullable(configMap.get(ConfigEnum.ICH_HIT_ALL)).ifPresent(value ->
            {
                this.ALL_HITS.reset();
                this.ALL_HITS.add(Long.parseLong(value));
            });
        } finally {
            lock.unlock();
        }
    }

    public void exportConfig(HashMap<ConfigEnum, String> config) {
        synchronized (config) {
            if (ALL_HITS.longValue() != INITIAL_CAPACITY) {
                config.put(ConfigEnum.ICH_HIT_ALL, JSON.toJSONString(ALL_HITS.longValue()));
            }
            if (!INCRE.isEmpty()) {
                config.put(ConfigEnum.ICH_HIT_MAP, JSON.toJSONString(INCRE));
            }
        }
    }
}